package org.example.heartbeat;

import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import org.apache.pekko.actor.*;
import org.apache.pekko.dispatch.OnComplete;
import org.apache.pekko.pattern.Patterns;
import scala.concurrent.ExecutionContext;
import scala.concurrent.Future;

import java.time.Duration;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executors;

public class HeartbeatService {
    public static class HeartbeatRequest implements HeartbeatMessage{
        static final HeartbeatRequest INSTANCE = new HeartbeatRequest();
        private HeartbeatRequest() {}
    }
    public static class HeartbeatResponse implements HeartbeatMessage {
        static final HeartbeatResponse INSTANCE = new HeartbeatResponse();
        private HeartbeatResponse() {}
    }
    public static class Worker extends AbstractActor {

        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(HeartbeatMessage.class , this::messageHandler)
                    .build();
        }

        private void messageHandler(HeartbeatMessage heartbeatMessage) {

            if (heartbeatMessage instanceof HeartbeatRequest){
                ActorSelection actorSelection = getContext().actorSelection("pekko://flink@127.0.0.1:17354/user/boss");
                actorSelection.tell(heartbeatMessage , getSelf());
//                Patterns.ask(actorSelection , heartbeatMessage , 1000l);
//                Patterns.ask(actorSelection, HeartbeatRequest.INSTANCE, 5000L).onComplete(
//                        new OnComplete<Object>() {
//                            @Override
//                            public void onComplete(Throwable failure, Object success) {
//                                if (failure != null) {
//                                    System.out.println("fa" + failure.toString());
//                                } else {
//                                    System.out.println("suc" + success);
//                                }
//                            }
//                        },DirectExecutionContext.INSTANCE
//                );
                System.out.println("worker" + heartbeatMessage.getClass().getName());
            }
//            getSender().tell(HeartbeatResponse.INSTANCE , ActorRef.noSender());
        }
    }
    public static class Boss extends AbstractActor{
        @Override
        public Receive createReceive() {
            return receiveBuilder()
                    .match(HeartbeatMessage.class , this::messageHandler)
                    .build();
        }

        private void messageHandler(HeartbeatMessage heartbeatMessage) {
            System.out.println("boss" + heartbeatMessage.getClass().getName());
        }
    }

    public static void main(String[] args) {
        Config load = ConfigFactory.load("application.conf");
        ActorSystem flink = ActorSystem.create("flink" , load);
        ActorRef boss = flink.actorOf(Props.create(Boss.class), "boss");
        ActorRef worker = flink.actorOf(Props.create(Boss.class), "worker");
        worker.tell(HeartbeatRequest.INSTANCE , ActorRef.noSender());
    }
    private static class DirectExecutionContext implements ExecutionContext {

        static final DirectExecutionContext INSTANCE = new DirectExecutionContext();

        private DirectExecutionContext() {}

        @Override
        public void execute(Runnable runnable) {
            runnable.run();
        }

        @Override
        public void reportFailure(Throwable cause) {
            throw new IllegalStateException("Error in direct execution context.", cause);
        }

        @Override
        public ExecutionContext prepare() {
            return this;
        }
    }
}
